package com.boarsoft.message.core;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.common.Util;
import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.message.MessageConfig;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.bean.MessageEntity;
import com.boarsoft.message.bean.MessageQueue;
import com.boarsoft.message.bean.RecipientMQ;
import com.boarsoft.message.bean.RecipientStatus;
import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.broker.RecipientFinder;
import com.boarsoft.message.store.MessageRoller;

/**
 * 消息发送API，默认通过使用RPC完成消息发送<br>
 * 在主线程中发送10个内存队列中的消息，在额外的一个线程中发送10个文件队列中的消息
 * 
 * @author Mac_J
 *
 */
// @Component("messageCore")
public class MessageCore implements MessageNode, MessageService, Runnable {
	private static final Logger log = LoggerFactory.getLogger(MessageCore.class);

	/** */
	@Autowired
	protected MessageContext messageContext;
	/** 手动注入远程消息中心消息服务 */
	protected MessageService messageService;

	protected Lock lock = new ReentrantLock();
	protected Condition condition = lock.newCondition();
	protected volatile boolean running = true;

	/** */
	protected ExecutorService threadPool;
	protected int threads = 1;

	@PostConstruct
	public void init() {
		if (threadPool == null) {
			threadPool = Executors.newCachedThreadPool();
		}
		for (int i = 0; i < threads; i++) {
			threadPool.execute(this);
		}
	}

	@Override
	public void put(Message m) throws Exception {
		// 打印上一个处理节点
		m.addTrace(MessageConfig.getAddr());
		// 获取队列（配置）本地队列与服务器上的同名队列配置是不同的
		TargetMQ tq = messageContext.getTargetQueue(m.getTarget());
		// 如果消息是从兄弟节点转来过的，就直接入队发送，反之正常处理
		if (m.isRelay()) {
			log.info("Handle message {} from sibling {}", m, m.lastTrace());
			m.setRelay(false);
			this.handle(tq, m); // 入队发送
			return;
		}
		// 如果还没有给ID就生成一个
		if (Util.strIsEmpty(m.getId())) {
			m.setId(RandomUtil.randomUUID());
		}
		log.debug("Handle message {} from {}", m, m.getFromAddr());
		// 检查队列的高可用策略，判断是否需要本地高可用
		switch (tq.getReliability()) {
		case TargetMQ.RA_MEM:
			// 不持久化，直接入队
			break;
		case TargetMQ.RA_TRAN:
		case TargetMQ.RA_SYNC:
			tq.getRoller().append(m); // 消息持久化，并入队
			return;
		case TargetMQ.RA_SYNC_MEM:
			tq.getRoller().append(m); // 消息持久化，但不入队
			m.setRelay(true); // 添加转发标识
			messageService.put(m); // 由兄弟节点发送
			return;
		default:
			throw new IllegalStateException(String.format(//
					"Unknown reliability %d for message %s", tq.getReliability(), m));
		}
		// 添加到队列，可能添加不成功
		this.handle(tq, m);
	}

	@Override
	public void run() {
		// 遍历targetQueueMap将消息按收件人分拣到对应的收件人队列
		Map<String, TargetMQ> quMap = messageContext.getTargetQueueMap();
		while (running) {
			try {
				boolean noMore = true;
				for (Entry<String, TargetMQ> en : quMap.entrySet()) {
					String target = en.getKey();
					TargetMQ tq = en.getValue();
					if (tq.isEmpty()) {
						// 移除为空的“已删除”队列
						if (target.startsWith("removed_")) {
							quMap.remove(target);
						}
						continue;
					}
					log.debug("Try to poll {}/{} messages from target queue {}", //
							tq.getPriority(), tq.size(), tq.getCode());
					for (int i = 0; i < tq.getPriority(); i++) {
						MessageEntity me = tq.poll();
						if (me == null) {
							break;
						}
						// 更新主消息表的尝试次数
						me.tried1();
						// 如果要发给远程消息服务器（生产者），就不入队，直接发送
						if (tq.isRelay() && messageService != null) {
							this.send(me, tq); // 发给消息服务器
						} else {
							this.dispatch(me, tq); // 入队，以便分发
						}
					}
					noMore = noMore && tq.isEmpty();
				}
				if (noMore) {
					this.await(quMap);// 此方法会再次判断
				}
				// 清除被移除的空目标队列和空收件人队列。
				messageContext.clear();
			} catch (Throwable e) {
				log.error("Error on handle target queues", e);
				try {
					Thread.sleep(10000L);
				} catch (InterruptedException e1) {
					// Nothing to do
				}
			}
		}
	}

	protected void send(MessageEntity me, TargetMQ tq) {
		log.debug("Send message {} to {}", me, messageService);
		Message m = me.getMessage();
		MessageRoller roller = tq.getRoller();
		try {
			messageService.put(m);
			if (roller == null) {
				return;
			}
			m.setStatus(Message.STATUS_SENT);
			roller.update(m, null);
			// roller.update(m.getId(), Message.STATUS_SENT, m.getTried(),
			// null);
		} catch (Exception e) {
			log.error("Error on send message {}", m, e);
			if (roller == null) {
				return;
			}
			try {
				m.setStatus(Message.STATUS_NEW);
				roller.update(m, e.getMessage());
				// roller.update(m.getId(), Message.STATUS_NEW, m.getTried(),
				// e.getMessage());
			} catch (Exception e1) {
				log.error("Error on update message {}", m, e1);
			}
		}
	}

	protected void dispatch(MessageEntity me, TargetMQ tq) {
		try {
			me.setStatus(Message.STATUS_SENDING, "dispatch");
			switch (tq.getType()) {
			case TargetMQ.TYPE_TOPIC:
				this.dispatchTopic(me, tq);
				break;
			case TargetMQ.TYPE_QUEUE:
				this.dispatchQueue(me, tq);
				break;
			case TargetMQ.TYPE_DYNAMIC:
				this.dispatchDynamic(me, tq);
				break;
			default:
				log.error("Type of message queue {} is unknown", tq.getType());
				break;
			}
		} catch (Exception e) {
			log.error("Error on dispatch message {}", me.getMessage().getId(), e);
		}
	}

	protected void dispatchTopic(MessageEntity me, MessageQueue mq) throws Exception {
		Message m = me.getMessage();
		TargetMQ tq = (TargetMQ) mq;
		Set<String> rs = tq.getRecipients();
		String v = Util.array2str(rs.toArray(), ";");
		log.debug("Get topic {} subscribers: {}", tq, v);
		if (rs.isEmpty()) {
			String s = String.format("No config of topic %s", tq.getCode());
			me.setStatus(Message.STATUS_INVALID, s);
			// this.onMessageDone(m, Message.STATUS_INVALID, s, tq);
			return;
		}
		//
		me.setStatus(Message.STATUS_SENDING);
		log.debug("Get topic subscribers, {} = {}", tq, v);
		for (String r : rs) {
			if (Util.strIsEmpty(r)) {
				log.warn("Empty subscriber of topic {}", tq);
				continue;
			}
			m.addRecipient(r);
		}
		for (Entry<String, RecipientStatus> en : m.getRecipients().entrySet()) {
			String r = en.getKey();
			log.debug("Put topic messsage {} to recipient queue {}", m, r);
			messageContext.getRecipientQueue(r).handle(me);
		}
	}

	protected void dispatchQueue(MessageEntity me, MessageQueue mq) throws Exception {
		Message m = me.getMessage();
		Set<Entry<String, RecipientStatus>> set = m.getRecipients().entrySet();
		if (set == null || set.isEmpty()) {
			TargetMQ tq = (TargetMQ) mq;
			Set<String> rl = tq.getRecipients();
			if (rl == null || rl.isEmpty()) {
				String s = String.format("No recipients for message %s", m.getId());
				log.error(s);
				me.setStatus(Message.STATUS_INVALID, s);
				return;
			}
			for (String r : rl) {
				m.addRecipient(r);
			}
		}
		me.setStatus(Message.STATUS_SENDING);
		for (Entry<String, RecipientStatus> en : set) {
			String r = en.getKey();
			log.debug("Put queue messsage {} to recipient queue {}", m, r);
			RecipientMQ rq = messageContext.getRecipientQueue(r);
			if (rq == null) {
				log.error("{} is invalid recipient", r);
				// me.setStatus(Message.STATUS_INVALID,//
				// String.format("{} is invalid recipient", r));
				continue;
			}
			rq.handle(me);
		}
	}

	protected void dispatchDynamic(MessageEntity me, MessageQueue mq) throws Exception {
		Message m = me.getMessage();
		TargetMQ tq = (TargetMQ) mq;
		RecipientFinder rf = tq.getRecipientFinder();
		if (rf == null) {
			log.error("Recipient finder {} does not exists", m.getTarget());
			String s = String.format("No recipient finder for dynamic MQ {}", tq.getCode());
			me.setStatus(Message.STATUS_INVALID, s);
			return;
		}
		//
		me.setStatus(Message.STATUS_SENDING);
		Map<String, Boolean> rm = rf.find(m);
		if (rm == null || rm.isEmpty()) {
			return; // 如果收件人列表为空，表示丢弃这条消息，不放入收件人队列
		}
		m.addRecipients(rm);
		for (Entry<String, Boolean> en : rm.entrySet()) {
			String r = en.getKey();
			log.debug("Put dynamic message {} to recipient queue {}", m, r);
			messageContext.getRecipientQueue(r).handle(me);
		}
	}

	public boolean handle(TargetMQ tq, Message m) throws Exception {
		if (!tq.add(m)) {
			log.warn("Can not add message {} because target queue {} is full", m, tq);
			return false;
		}
		lock.lock();
		try {
			condition.signalAll();
		} finally {
			lock.unlock();
		}
		return true;
	}

	public boolean handle(TargetMQ tq, MessageEntity me) throws Exception {
		if (!tq.add(me)) {
			return false;
		}
		lock.lock();
		try {
			condition.signalAll();
		} finally {
			lock.unlock();
		}
		return true;
	}

	protected void await(Map<String, TargetMQ> quMap) {
		lock.lock();
		try {
			for (Entry<String, TargetMQ> en : quMap.entrySet()) {
				TargetMQ tq = en.getValue();
				if (!tq.isEmpty()) {
					return;
				}
			}
			try {
				condition.await(1000L, TimeUnit.MILLISECONDS);
			} catch (Exception e) {
				log.warn("Be interrupted while awaiting", e);
			}
		} finally {
			lock.unlock();
		}
	}

	@PreDestroy
	public void destroy() {
		log.warn("Message core is stopping");
		running = false;
	}

	public boolean isRunning() {
		return running;
	}

	public void updateProp(Map<String, String> propMap) throws FileNotFoundException, IOException, URISyntaxException {
		for (Entry<String, String> en : propMap.entrySet()) {
			String k = en.getKey();
			String v = en.getValue();
			log.warn("Message config {}={} be changed to {}", k, MessageConfig.get(k), v);
			MessageConfig.set(k, v);
		}
		MessageConfig.flush();
	}

	// -------------

	public MessageService getMessageService() {
		return messageService;
	}

	public void setMessageService(MessageService messageService) {
		this.messageService = messageService;
	}

	public MessageContext getMessageContext() {
		return messageContext;
	}

	public void setMessageContext(MessageContext messageContext) {
		this.messageContext = messageContext;
	}

	public ExecutorService getThreadPool() {
		return threadPool;
	}

	public void setThreadPool(ExecutorService threadPool) {
		this.threadPool = threadPool;
	}

	public int getThreads() {
		return threads;
	}

	public void setThreads(int threads) {
		this.threads = threads;
	}

}
